"""
第二课：异步编程实战练习代码
配合 02_async_programming.md 讲义使用

运行方式：
cd /Users/baimu/PycharmProjects/fastApiProject3
python lessons/02_async_programming_practice.py
"""

import asyncio
import time
from typing import List, Dict, Any

print("=" * 60)
print("第一部分：同步 vs 异步对比")
print("=" * 60)


# 1. 同步版本（慢）
def sync_task(name: str, duration: int) -> str:
    """同步任务"""
    print(f"[同步] 任务 {name} 开始...")
    time.sleep(duration)
    print(f"[同步] 任务 {name} 完成！")
    return f"结果-{name}"


def sync_main():
    """同步执行多个任务"""
    print("\n▶ 同步执行示例:")
    start = time.time()

    result1 = sync_task("A", 2)
    result2 = sync_task("B", 1)
    result3 = sync_task("C", 1)

    elapsed = time.time() - start
    print(f"✓ 总耗时: {elapsed:.1f}秒")
    print(f"✓ 结果: [{result1}, {result2}, {result3}]\n")


sync_main()


# 2. 异步版本（快）
async def async_task(name: str, duration: int) -> str:
    """异步任务"""
    print(f"[异步] 任务 {name} 开始...")
    await asyncio.sleep(duration)
    print(f"[异步] 任务 {name} 完成！")
    return f"结果-{name}"


async def async_main():
    """异步并发执行多个任务"""
    print("▶ 异步执行示例:")
    start = time.time()

    # 方法1：使用 create_task
    # task1 = asyncio.create_task(async_task("A", 2))
    # task2 = asyncio.create_task(async_task("B", 1))
    # task3 = asyncio.create_task(async_task("C", 1))
    # results = [await task1, await task2, await task3]

    # 方法2：使用 gather（更简洁）
    results = await asyncio.gather(
        async_task("A", 2), async_task("B", 1), async_task("C", 1)
    )

    elapsed = time.time() - start
    print(f"✓ 总耗时: {elapsed:.1f}秒")
    print(f"✓ 结果: {results}\n")


asyncio.run(async_main())

print("💡 对比：异步版本节省了约2秒！")

print("\n" + "=" * 60)
print("第二部分：async/await 基础")
print("=" * 60)


# 3. 基础异步函数
async def greet(name: str) -> str:
    """异步问候函数"""
    await asyncio.sleep(0.5)  # 模拟异步操作
    return f"你好，{name}！"


async def demo_basic_async():
    """演示基础异步函数"""
    print("\n▶ 基础异步函数:")

    # 串行调用
    result1 = await greet("张三")
    result2 = await greet("李四")
    print(f"串行结果: {result1}, {result2}")

    # 并行调用
    results = await asyncio.gather(greet("王五"), greet("赵六"))
    print(f"并行结果: {results}\n")


asyncio.run(demo_basic_async())


# 4. 异步数据获取
async def fetch_user(user_id: int) -> Dict[str, Any]:
    """模拟从数据库获取用户"""
    print(f"  → 开始获取用户 {user_id}")
    await asyncio.sleep(1)  # 模拟数据库查询延迟
    print(f"  ← 用户 {user_id} 获取完成")
    return {"id": user_id, "name": f"User{user_id}", "age": 20 + user_id}


async def demo_fetch_users():
    """演示并发获取多个用户"""
    print("▶ 并发获取用户数据:")
    start = time.time()

    # 并发获取5个用户
    user_ids = [1, 2, 3, 4, 5]
    users = await asyncio.gather(*[fetch_user(uid) for uid in user_ids])

    elapsed = time.time() - start
    print(f"\n✓ 获取了 {len(users)} 个用户，耗时 {elapsed:.1f}秒")
    for user in users:
        print(f"  - {user}")
    print()


asyncio.run(demo_fetch_users())

print("=" * 60)
print("第三部分：异步并发模式")
print("=" * 60)


# 5. create_task 创建任务
async def demo_create_task():
    """演示 create_task 的使用"""
    print("\n▶ create_task 示例:")
    start = time.time()

    # 创建任务（立即开始执行）
    task1 = asyncio.create_task(async_task("Task-1", 2))
    task2 = asyncio.create_task(async_task("Task-2", 1))

    print("任务已创建，正在后台运行...")

    # 等待任务完成
    result1 = await task1
    result2 = await task2

    elapsed = time.time() - start
    print(f"✓ 两个任务完成，耗时 {elapsed:.1f}秒\n")


asyncio.run(demo_create_task())


# 6. gather 批量执行
async def demo_gather():
    """演示 gather 的使用"""
    print("▶ gather 批量执行:")
    start = time.time()

    # 一次性启动多个任务
    results = await asyncio.gather(
        async_task("G-1", 1),
        async_task("G-2", 2),
        async_task("G-3", 1),
        async_task("G-4", 1),
    )

    elapsed = time.time() - start
    print(f"✓ 所有任务完成，耗时 {elapsed:.1f}秒")
    print(f"✓ 结果: {results}\n")


asyncio.run(demo_gather())

print("=" * 60)
print("第四部分：实战场景")
print("=" * 60)


# 7. 场景1：早餐准备
async def make_coffee() -> str:
    """煮咖啡"""
    print("  ☕ 开始煮咖啡...")
    await asyncio.sleep(3)
    print("  ☕ 咖啡煮好了！")
    return "咖啡"


async def make_toast() -> str:
    """烤面包"""
    print("  🍞 开始烤面包...")
    await asyncio.sleep(2)
    print("  🍞 面包烤好了！")
    return "面包"


async def fry_eggs() -> str:
    """煎鸡蛋"""
    print("  🍳 开始煎鸡蛋...")
    await asyncio.sleep(2)
    print("  🍳 鸡蛋煎好了！")
    return "鸡蛋"


async def prepare_breakfast():
    """准备早餐"""
    print("\n▶ 场景1：准备早餐")
    print("开始准备早餐...")
    start = time.time()

    # 并发准备所有食物
    breakfast = await asyncio.gather(make_coffee(), make_toast(), fry_eggs())

    elapsed = time.time() - start
    print(f"\n✓ 早餐准备完毕: {', '.join(breakfast)}")
    print(f"✓ 总耗时: {elapsed:.1f}秒\n")


asyncio.run(prepare_breakfast())


# 8. 场景2：批量处理订单
class Order:
    """订单类"""

    def __init__(self, order_id: int, items: List[str]):
        self.order_id = order_id
        self.items = items
        self.status = "pending"


async def process_order(order: Order) -> Order:
    """处理订单"""
    print(f"  📦 开始处理订单 #{order.order_id}")
    await asyncio.sleep(1)  # 模拟处理时间
    order.status = "completed"
    print(f"  ✓ 订单 #{order.order_id} 处理完成")
    return order


async def batch_process_orders():
    """批量处理订单"""
    print("▶ 场景2：批量处理订单")

    # 创建10个订单
    orders = [Order(i, [f"商品{i}"]) for i in range(1, 11)]

    print(f"收到 {len(orders)} 个订单，开始处理...\n")
    start = time.time()

    # 并发处理所有订单
    processed_orders = await asyncio.gather(*[process_order(order) for order in orders])

    elapsed = time.time() - start
    completed = sum(1 for o in processed_orders if o.status == "completed")

    print(f"\n✓ 处理完成: {completed}/{len(orders)} 个订单")
    print(f"✓ 总耗时: {elapsed:.1f}秒")
    print(f"💡 如果串行处理需要: {len(orders)}秒\n")


asyncio.run(batch_process_orders())


# 9. 场景3：爬虫（模拟）
async def fetch_page(url: str, delay: float) -> Dict[str, Any]:
    """模拟抓取网页"""
    print(f"  🌐 开始抓取: {url}")
    await asyncio.sleep(delay)  # 模拟网络延迟
    content = f"<html>Content from {url}</html>"
    print(f"  ✓ 抓取完成: {url}")
    return {"url": url, "content": content, "length": len(content)}


async def web_crawler():
    """网页爬虫"""
    print("▶ 场景3：网页爬虫")

    urls = [
        "https://example.com/page1",
        "https://example.com/page2",
        "https://example.com/page3",
        "https://example.com/page4",
        "https://example.com/page5",
    ]

    print(f"开始抓取 {len(urls)} 个网页...\n")
    start = time.time()

    # 并发抓取所有页面
    results = await asyncio.gather(*[fetch_page(url, 1.5) for url in urls])

    elapsed = time.time() - start
    print(f"\n✓ 抓取完成，共获取 {len(results)} 个页面")
    print(f"✓ 总耗时: {elapsed:.1f}秒")
    print(f"💡 串行抓取需要: {len(urls) * 1.5:.1f}秒\n")


asyncio.run(web_crawler())

print("=" * 60)
print("第五部分：错误处理")
print("=" * 60)


# 10. 异步函数中的异常处理
async def risky_task(task_id: int, should_fail: bool = False) -> str:
    """可能失败的任务"""
    print(f"  ⚙ 任务 {task_id} 开始")
    await asyncio.sleep(0.5)

    if should_fail:
        raise ValueError(f"任务 {task_id} 失败！")

    print(f"  ✓ 任务 {task_id} 成功")
    return f"结果-{task_id}"


async def demo_error_handling():
    """演示错误处理"""
    print("\n▶ 异常处理示例:")

    # 方法1：try/except
    print("\n1. 单个任务的异常处理:")
    try:
        await risky_task(1, should_fail=True)
    except ValueError as e:
        print(f"  ❌ 捕获异常: {e}")

    # 方法2：gather 中的异常
    print("\n2. gather 中的异常处理:")
    try:
        results = await asyncio.gather(
            risky_task(2, should_fail=False),
            risky_task(3, should_fail=True),  # 这个会失败
            risky_task(4, should_fail=False),
        )
    except ValueError as e:
        print(f"  ❌ gather 捕获异常: {e}")

    # 方法3：gather 的 return_exceptions 参数
    print("\n3. gather 返回异常（不中断）:")
    results = await asyncio.gather(
        risky_task(5, should_fail=False),
        risky_task(6, should_fail=True),
        risky_task(7, should_fail=False),
        return_exceptions=True,  # 不抛出异常，而是返回
    )

    print(f"\n  结果:")
    for i, result in enumerate(results, 5):
        if isinstance(result, Exception):
            print(f"    任务 {i}: ❌ {result}")
        else:
            print(f"    任务 {i}: ✓ {result}")
    print()


asyncio.run(demo_error_handling())

print("=" * 60)
print("第六部分：超时控制")
print("=" * 60)


# 11. 超时控制
async def slow_task(duration: int) -> str:
    """慢速任务"""
    print(f"  ⏳ 慢速任务开始（需要 {duration} 秒）")
    await asyncio.sleep(duration)
    print(f"  ✓ 慢速任务完成")
    return "完成"


async def demo_timeout():
    """演示超时控制"""
    print("\n▶ 超时控制示例:")

    # 案例1：任务在超时前完成
    print("\n1. 任务按时完成:")
    try:
        result = await asyncio.wait_for(slow_task(2), timeout=3)
        print(f"  ✓ 结果: {result}")
    except asyncio.TimeoutError:
        print(f"  ❌ 超时！")

    # 案例2：任务超时
    print("\n2. 任务超时:")
    try:
        result = await asyncio.wait_for(slow_task(5), timeout=2)
        print(f"  ✓ 结果: {result}")
    except asyncio.TimeoutError:
        print(f"  ❌ 任务超过 2 秒，已取消！")
    print()


asyncio.run(demo_timeout())

print("=" * 60)
print("第七部分：实战练习")
print("=" * 60)


# 练习1：异步计算器
async def async_add(a: int, b: int) -> int:
    """异步加法"""
    await asyncio.sleep(0.1)
    return a + b


async def async_multiply(a: int, b: int) -> int:
    """异步乘法"""
    await asyncio.sleep(0.1)
    return a * b


async def practice1_calculator():
    """练习1：并发计算多个表达式"""
    print("\n▶ 练习1：异步计算器")

    # 并发计算多个表达式
    results = await asyncio.gather(
        async_add(10, 20), async_multiply(5, 6), async_add(100, 200)
    )

    print(f"✓ 计算结果: 10+20={results[0]}, 5×6={results[1]}, 100+200={results[2]}\n")


asyncio.run(practice1_calculator())


# 练习2：模拟下载文件
async def download_file(filename: str, size_mb: int) -> Dict[str, Any]:
    """模拟下载文件"""
    print(f"  📥 开始下载: {filename} ({size_mb}MB)")
    await asyncio.sleep(size_mb * 0.5)  # 假设每MB需要0.5秒
    print(f"  ✓ 下载完成: {filename}")
    return {"filename": filename, "size": size_mb, "status": "completed"}


async def practice2_download_files():
    """练习2：并发下载多个文件"""
    print("▶ 练习2：并发下载文件")

    files = [
        ("document.pdf", 5),
        ("video.mp4", 10),
        ("image.jpg", 2),
        ("music.mp3", 3),
    ]

    start = time.time()

    # 并发下载所有文件
    results = await asyncio.gather(*[download_file(name, size) for name, size in files])

    elapsed = time.time() - start
    total_size = sum(r["size"] for r in results)

    print(f"\n✓ 所有文件下载完成")
    print(f"✓ 总大小: {total_size}MB，总耗时: {elapsed:.1f}秒")
    print(f"💡 串行下载需要: {sum(size * 0.5 for _, size in files):.1f}秒\n")


asyncio.run(practice2_download_files())


# 练习3：异步生产者-消费者
async def producer(queue: asyncio.Queue, producer_id: int):
    """生产者"""
    for i in range(3):
        item = f"P{producer_id}-Item{i+1}"
        await asyncio.sleep(0.5)
        await queue.put(item)
        print(f"  ➕ 生产者 {producer_id} 生产: {item}")
    print(f"  ✓ 生产者 {producer_id} 完成")


async def consumer(queue: asyncio.Queue, consumer_id: int, total_items: int):
    """消费者"""
    for _ in range(total_items):
        item = await queue.get()
        await asyncio.sleep(0.3)
        print(f"  ➖ 消费者 {consumer_id} 消费: {item}")
    print(f"  ✓ 消费者 {consumer_id} 完成")


async def practice3_producer_consumer():
    """练习3：生产者-消费者模式"""
    print("▶ 练习3：生产者-消费者模式")

    queue = asyncio.Queue()

    # 2个生产者，每个生产3个项目 = 总共6个项目
    # 3个消费者，每个消费2个项目
    await asyncio.gather(
        producer(queue, 1),
        producer(queue, 2),
        consumer(queue, 1, 2),
        consumer(queue, 2, 2),
        consumer(queue, 3, 2),
    )

    print(f"\n✓ 生产-消费完成\n")


asyncio.run(practice3_producer_consumer())

print("=" * 60)
print("🎉 恭喜！你已完成异步编程的所有练习")
print("=" * 60)
print("\n核心要点回顾:")
print("1. async def 定义异步函数，await 等待结果")
print("2. 使用 asyncio.gather() 实现并发执行")
print("3. 使用 asyncio.create_task() 创建后台任务")
print("4. 异步编程可以显著提升 I/O 密集型任务的性能")
print("5. FastAPI 完美支持异步编程")
print("\n准备好继续学习「HTTP 协议与 RESTful API」了吗？")

